package com.smsc.headend.connector.config;


import com.google.common.collect.Maps;
import com.smsc.headend.common.thread.UDISThreadFactory;
import com.smsc.headend.connector.manager.ConnectorTaskThread;
import com.smsc.headend.connector.manager.TaskTicketManager;
import com.smsc.headend.connector.service.DeviceConnectService;
import com.smsc.headend.module.task.dto.ConnectorTask;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.annotation.Order;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;

@Configuration
@EnableKafka
@Slf4j
@Order(1)
public class KafkaConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private Boolean autoCommit;

    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;

    @Value("${spring.kafka.consumer.max-poll-records}")
    private Integer maxPollRecords;

    @Value("${spring.kafka.producer.linger}")
    private int linger;

    @Value("${spring.kafka.producer.retries}")
    private Integer retries;

    @Value("${spring.kafka.producer.batch-size}")
    private Integer batchSize;

    @Value("${spring.kafka.producer.buffer-memory}")
    private Integer bufferMemory;

    @Value("${packet.thread.coreSize}")
    Integer coreSize;
    @Value("${packet.thread.maxPoolSize}")
    Integer maxPoolSize;
    @Value("${packet.thread.aliveTime}")
    Integer aliveTime;

    @Value("${connector.thread.coreSize}")
    Integer connectorTaskSize;
    @Value("${connector.thread.maxPoolSize}")
    Integer connectorMaxPoolSize;
    @Value("${connector.thread.aliveTime}")
    Integer connectorAliveTime;
    @Autowired
    DeviceConnectService deviceConnectService;
    @Autowired
    @Lazy
    TaskTicketManager taskTicketManager;

    @Autowired
    DiscoveryClient discoveryClient;

    //cankao :https://blog.csdn.net/tmeng521/article/details/90901925
    public Map<String, Object> producerConfigs() {

        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        //设置重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        //达到batchSize大小的时候会发送消息
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        //延时时间，延时时间到达之后计算批量发送的大小没达到也发送消息
        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
        //缓冲区的值
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        //序列化手段
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        //producer端的消息确认机制,-1和all都表示消息不仅要写入本地的leader中还要写入对应的副本中
        props.put(ProducerConfig.ACKS_CONFIG, "-1");//单个brok 推荐使用'1'
        //单条消息的最大值以字节为单位,默认值为1048576
//        props.put(ProducerConfig.LINGER_MS_CONFIG, 10485760);
        //设置broker响应时间，如果broker在60秒之内还是没有返回给producer确认消息，则认为发送失败
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
        //指定拦截器(value为对应的class)
        //props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.te.handler.KafkaProducerInterceptor");
        //设置压缩算法(默认是木有压缩算法的)
//        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");//snappy
        return props;
    }


    @Bean //创建一个kafka管理类，相当于rabbitMQ的管理类rabbitAdmin,没有此bean无法自定义的使用adminClient创建topic
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> props = new HashMap<>();
        //配置Kafka实例的连接地址
        //kafka的地址，不是zookeeper
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        return new KafkaAdmin(props);
    }

    @Bean("savePacketPool")
    public ThreadPoolExecutor savePacketPool() {
        return new ThreadPoolExecutor(coreSize, maxPoolSize, aliveTime, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new UDISThreadFactory());
    }

    @Bean("connectorTaskPool")
    public ThreadPoolExecutor connectorTaskPool() {
        return new ThreadPoolExecutor(connectorTaskSize, connectorMaxPoolSize, connectorAliveTime, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000), new UDISThreadFactory("ute-task"), (r, executor) -> {
            log.warn("ConnectorTaskPoolRejecting");
            if (r.getClass().isAssignableFrom(ConnectorTaskThread.class)) {
                try {
                    ConnectorTaskThread connectorTaskThread = (ConnectorTaskThread) r;
                    taskTicketManager.recycleTicket(connectorTaskThread.getTicketNumber());
                    ConnectorTask connectorTask = connectorTaskThread.getConnectorTask();
                    connectorTask.setTicketNumber(null);
                    log.warn("recycleTicketAndEnterTaskQueue", connectorTask.getTaskNo());
                    taskTicketManager.addTaskToQueue(connectorTask);
                } catch (Exception e) {
                    log.error("reenter task failed", e);
                }
            }
        });
    }

    @Bean("connectorReceiveTaskPool")
    public ThreadPoolExecutor connectorReceiveTaskPool() {
        return new ThreadPoolExecutor(connectorTaskSize, connectorMaxPoolSize, connectorAliveTime, TimeUnit.SECONDS, new LinkedBlockingQueue<>(5000), new UDISThreadFactory("device"), (r, executor) -> {
            log.warn("receivePacketOverlimit,discard");
        });
    }

    @Bean("retryService")
    public ScheduledExecutorService buildRetryServicePool() {
        ScheduledExecutorService scheduledThreadPoolExecutor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new UDISThreadFactory("retry"));
        return scheduledThreadPoolExecutor;
    }

    @Bean  //kafka客户端，在spring中创建这个bean之后可以注入并且创建topic,用于集群环境，创建对个副本
    public AdminClient adminClient() {
        return AdminClient.create(kafkaAdmin().getConfig());
    }


    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }


    @Bean("consumerConfigs")
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = Maps.newHashMap();
        props.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString().replace("-", ""));
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
//        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 180000);
//        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 900000);
//        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 900000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
        return props;
    }

    public Map<String, Object> connectorConfigs() {
        Map<String, Object> props = Maps.newHashMap();
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "connector");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
//        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 180000);
//        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 900000);
//        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 900000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
        return props;
    }


    @Bean("connectBatchFactory")
    @Order(0)
    public KafkaListenerContainerFactory<?> batchConnectorFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(connectorConfigs()));
        //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(true);
        // set the retry template
//        factory.setRetryTemplate(retryTemplate());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }

    @Bean("batchFactory")
    @Order(0)
    public KafkaListenerContainerFactory<?> batchFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(true);
        // set the retry template
//        factory.setRetryTemplate(retryTemplate());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
//        factory.setConcurrency(4);
        return factory;
    }
//    @KafkaListener(containerFactory = "batchFactory", topicPartitions = @TopicPartition(topic = TaskKafkaTopic.TASK_SEND_TO_CON
//            , partitions = "#{@finder.partitions('" + TaskKafkaTopic.TASK_SEND_TO_CON + "')}"))


    @Bean
    public PartitionFinder finder(ConsumerFactory<String, String> consumerFactory) {
        return new PartitionFinder(consumerFactory);
    }

    public static class PartitionFinder {
        private final ConsumerFactory<String, String> consumerFactory;

        public PartitionFinder(ConsumerFactory<String, String> consumerFactory) {
            this.consumerFactory = consumerFactory;
        }


        public String[] partitions(String topic) {
//            DiscoveryClient discoveryClient = SpringContextUtils.getBean(DiscoveryClient.class);
//            ConcurrentKafkaListenerContainerFactory factory = (ConcurrentKafkaListenerContainerFactory) SpringContextUtils.getBean("batchFactory");

//            System.out.println(discoveryClient.getOrder());
            try (Consumer<String, String> consumer = this.consumerFactory.createConsumer()) {
                return consumer.partitionsFor(topic).stream()
                        .map(pi -> "" + pi.partition())
                        .toArray(String[]::new);
            }
        }
    }

}
